package com.flink.hbase.sql2hdfs.udf;

import org.apache.flink.table.functions.ScalarFunction;

import java.time.LocalDateTime;
import java.time.format.DateTimeFormatter;
import java.util.regex.Pattern;

/**
 * 数据处理UDF
 * 包含时间处理、数据清洗、格式化等功能
 */
public class DataProcessUDF extends ScalarFunction {
    private static final long serialVersionUID = 1L;
    
    private static final DateTimeFormatter DEFAULT_FORMATTER = DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss");
    private static final DateTimeFormatter DATE_FORMATTER = DateTimeFormatter.ofPattern("yyyy-MM-dd");
    
    // 正则表达式模式
    private static final Pattern EMAIL_PATTERN = Pattern.compile(
        "^[a-zA-Z0-9_+&*-]+(?:\\.[a-zA-Z0-9_+&*-]+)*@(?:[a-zA-Z0-9-]+\\.)+[a-zA-Z]{2,7}$"
    );
    private static final Pattern PHONE_PATTERN = Pattern.compile(
        "^1[3-9]\\d{9}$"
    );
    
    /**
     * 获取当前处理时间
     * @return 当前时间字符串
     */
    public String getCurrentProcessTime() {
        return LocalDateTime.now().format(DEFAULT_FORMATTER);
    }
    
    /**
     * 获取当前日期作为分区日期
     * @return 当前日期字符串
     */
    public String getCurrentPartitionDate() {
        return LocalDateTime.now().format(DATE_FORMATTER);
    }
    
    /**
     * 格式化时间为指定格式
     * @param dateTime 时间
     * @param pattern 格式模式
     * @return 格式化后的时间字符串
     */
    public String formatDateTime(LocalDateTime dateTime, String pattern) {
        if (dateTime == null) {
            return null;
        }
        
        try {
            DateTimeFormatter formatter = DateTimeFormatter.ofPattern(pattern);
            return dateTime.format(formatter);
        } catch (Exception e) {
            return dateTime.format(DEFAULT_FORMATTER);
        }
    }
    
    /**
     * 格式化时间为默认格式
     * @param dateTime 时间
     * @return 格式化后的时间字符串
     */
    public String formatDateTime(LocalDateTime dateTime) {
        return formatDateTime(dateTime, "yyyy-MM-dd HH:mm:ss");
    }
    
    /**
     * 清洗用户名（去除特殊字符，保留中英文和数字）
     * @param userName 用户名
     * @return 清洗后的用户名
     */
    public String cleanUserName(String userName) {
        if (userName == null || userName.trim().isEmpty()) {
            return "未知用户";
        }
        
        // 去除前后空格，保留中文、英文、数字
        String cleaned = userName.trim().replaceAll("[^\\u4e00-\\u9fa5a-zA-Z0-9]", "");
        
        if (cleaned.isEmpty()) {
            return "用户" + userName.hashCode();
        }
        
        // 限制长度
        if (cleaned.length() > 50) {
            cleaned = cleaned.substring(0, 50);
        }
        
        return cleaned;
    }
    
    /**
     * 验证邮箱格式
     * @param email 邮箱
     * @return 是否为有效邮箱
     */
    public Boolean isValidEmail(String email) {
        if (email == null || email.trim().isEmpty()) {
            return false;
        }
        return EMAIL_PATTERN.matcher(email.trim()).matches();
    }
    
    /**
     * 清洗邮箱
     * @param email 邮箱
     * @return 清洗后的邮箱或null
     */
    public String cleanEmail(String email) {
        if (email == null || email.trim().isEmpty()) {
            return null;
        }
        
        String cleaned = email.trim().toLowerCase();
        return isValidEmail(cleaned) ? cleaned : null;
    }
    
    /**
     * 验证手机号格式
     * @param phone 手机号
     * @return 是否为有效手机号
     */
    public Boolean isValidPhone(String phone) {
        if (phone == null || phone.trim().isEmpty()) {
            return false;
        }
        
        String cleaned = phone.trim().replaceAll("[^0-9]", "");
        return PHONE_PATTERN.matcher(cleaned).matches();
    }
    
    /**
     * 清洗手机号
     * @param phone 手机号
     * @return 清洗后的手机号或null
     */
    public String cleanPhone(String phone) {
        if (phone == null || phone.trim().isEmpty()) {
            return null;
        }
        
        String cleaned = phone.trim().replaceAll("[^0-9]", "");
        return isValidPhone(cleaned) ? cleaned : null;
    }
    
    /**
     * 脱敏手机号（中间4位用*号替换）
     * @param phone 手机号
     * @return 脱敏后的手机号
     */
    public String maskPhone(String phone) {
        if (!isValidPhone(phone)) {
            return phone;
        }
        
        String cleaned = cleanPhone(phone);
        if (cleaned != null && cleaned.length() == 11) {
            return cleaned.substring(0, 3) + "****" + cleaned.substring(7);
        }
        return phone;
    }
    
    /**
     * 脱敏邮箱（用户名部分用*号替换）
     * @param email 邮箱
     * @return 脱敏后的邮箱
     */
    public String maskEmail(String email) {
        if (!isValidEmail(email)) {
            return email;
        }
        
        String cleaned = cleanEmail(email);
        if (cleaned != null) {
            int atIndex = cleaned.indexOf('@');
            if (atIndex > 0) {
                String username = cleaned.substring(0, atIndex);
                String domain = cleaned.substring(atIndex);
                
                if (username.length() <= 2) {
                    return "*" + domain;
                } else {
                    return username.charAt(0) + "***" + username.charAt(username.length() - 1) + domain;
                }
            }
        }
        return email;
    }
    
    /**
     * 清洗地址信息
     * @param address 地址
     * @return 清洗后的地址
     */
    public String cleanAddress(String address) {
        if (address == null || address.trim().isEmpty()) {
            return null;
        }
        
        String cleaned = address.trim();
        
        // 去除多余的空格
        cleaned = cleaned.replaceAll("\\s+", " ");
        
        // 限制长度
        if (cleaned.length() > 200) {
            cleaned = cleaned.substring(0, 200);
        }
        
        return cleaned;
    }
    
    /**
     * 验证年龄范围
     * @param age 年龄
     * @return 是否为有效年龄
     */
    public Boolean isValidAge(Integer age) {
        return age != null && age >= 0 && age <= 120;
    }
    
    /**
     * 清洗年龄
     * @param age 年龄
     * @return 清洗后的年龄或null
     */
    public Integer cleanAge(Integer age) {
        return isValidAge(age) ? age : null;
    }
    
    /**
     * 验证薪资范围
     * @param salary 薪资
     * @return 是否为有效薪资
     */
    public Boolean isValidSalary(Double salary) {
        return salary != null && salary >= 0 && salary <= 1000000;
    }
    
    /**
     * 清洗薪资
     * @param salary 薪资
     * @return 清洗后的薪资或null
     */
    public Double cleanSalary(Double salary) {
        return isValidSalary(salary) ? salary : null;
    }
    
    /**
     * 生成数据质量评分
     * @param hasValidAge 是否有有效年龄
     * @param hasValidEmail 是否有有效邮箱
     * @param hasValidPhone 是否有有效手机号
     * @param hasValidSalary 是否有有效薪资
     * @return 数据质量评分（0-100）
     */
    public Integer calculateDataQuality(Boolean hasValidAge, Boolean hasValidEmail, 
                                      Boolean hasValidPhone, Boolean hasValidSalary) {
        int score = 0;
        
        if (hasValidAge != null && hasValidAge) score += 25;
        if (hasValidEmail != null && hasValidEmail) score += 25;
        if (hasValidPhone != null && hasValidPhone) score += 25;
        if (hasValidSalary != null && hasValidSalary) score += 25;
        
        return score;
    }
    
    /**
     * 生成唯一ID
     * @param prefix 前缀
     * @param value 值
     * @return 唯一ID
     */
    public String generateId(String prefix, String value) {
        if (prefix == null) prefix = "ID";
        if (value == null) value = "";
        
        long hash = (prefix + value + System.currentTimeMillis()).hashCode();
        return prefix + "_" + Math.abs(hash);
    }
} 